﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using NetMicro.Cache;
using NetMicro.Core.Extensions;
using NetMicro.EventBus.Extensions;
using NetMicro.EventBus.Messages;
using NetMicro.EventBus.Persistences;
using NetMicro.EventBus.Policies;

namespace NetMicro.EventBus.BackgroundServices
{
    public class EventDelayMessageServiceProvider : IEventDelayMessageServiceProvider
    {
        private IOptions<EventBusOptions> Options { get; }
        private IRedisCache RedisCache { get; }

        private IMessageSerializer MessageSerializer { get; }

        private ILogger<EventDelayMessageServiceProvider> Logger { get; }

        private IDelayMessageStorage DelayMessageStorage { get; }
        private IMessageSender MessageSender { get; }


        public EventDelayMessageServiceProvider(IOptions<EventBusOptions> options,
            IRedisCache redisCache,
            IMessageSerializer messageSerializer,
            ILogger<EventDelayMessageServiceProvider> logger,
            IEventErrorHandler eventErrorHandler,
            IDelayMessageStorage delayMessageStorage,
            IMessageSender messageSender)
        {
            Options = options;
            RedisCache = redisCache;
            MessageSerializer = messageSerializer;
            Logger = logger;
            DelayMessageStorage = delayMessageStorage;
            MessageSender = messageSender;
        }
        public Task Execute(CancellationToken cancellationToken)
        {
            Task.Factory.StartNew(async () =>
            {
                await Handler();
            }, TaskCreationOptions.LongRunning);
            return Task.CompletedTask;
        }

        public async Task Handler()
        {

            while (true)
            {
                try
                {
                    var delayMessage = await DelayMessageStorage.GetMessagesAsync();
                    if (!delayMessage.Any())
                    {
                        await Task.Delay(TimeSpan.FromSeconds(2));
                        continue;
                    }
                    foreach (var item in delayMessage)
                    {
                        if (item.DelayAt > TimeSpan.Zero.GetTimestampSeconds())
                            continue;
                        string key = $"{EventBusConsts.EventBusTopic}:{item.Id}";
                        bool flag = await RedisCache.SetAsync(key, item.Id, 60);

                        if (flag)
                        {
                            try
                            {
                                await DelayMessageStorage.LockAsync(item.Id);
                                MessageTransfer messageTransfer = MessageSerializer.ConvertMessageTransfer(item);
                                await MessageSender.Send(messageTransfer);
                                await DelayMessageStorage.DeleteMessageAsync(item.Id);
                            }
                            catch (Exception ex)
                            {
                                await DelayMessageStorage.UnLockAsync(item.Id);
                                Logger.LogError(ex, "延迟队列执行发布MQ异常：");
                            }
                        }
                    }

                }
                catch (Exception ex)
                {

                    Logger.LogError(ex, "延迟队列执行发布异常：");
                    await Task.Delay(500);
                }
            }
        }
    }
}
